import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple3;

import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.Paths;

import static org.apache.flink.api.java.aggregation.Aggregations.*;

public class AggregateonfullTupleDataSet {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).append("/dataset_api/Java/tupledata.csv").toString();
        Path path = Paths.get(rootPath);
        boolean exists = Files.exists(path, new LinkOption[]{LinkOption.NOFOLLOW_LINKS});
        System.out.println("----：" + rootPath + "---"+ exists);
        DataSet<Tuple2<Integer, Double>> input = env.readCsvFile("file:///" + rootPath ).types(Integer.class, Double.class);


        input.print();
        System.out.println("--------------------------------------");
        DataSet<Tuple2<Integer, Double>> output = input
                .aggregate(SUM, 0)    // compute sum of the first field
                .and(MAX, 1); // compute minimum of the second field
//                .and(MAX,2);


        output.print();
//        第0个key求和,
//        第1个key求最小值
    }

}
